<?php

/**
 * @author superx
 */
abstract class My_RedisTask extends My_Task {

    public static $connName = 'default';
    public static $queueName = '';

    public $blockTimeout = 60;
    public $retryOnFailure = true;
    public $fetchRows = 1;

    public function run() {
        while (1) {
            $this->updateStatus();
            $job = $this->fetchJob();
            if (!$job) {
                //nothing to do
                $this->println('no job for now...');
                sleep(5);
            } else {
                try {
                    call_user_func([$this, 'worker'], $job);
                } catch (Exception $e) {
                    $this->println("job failed " . $e->getTraceAsString());
                    $this->redoJob($job);
                }
            }
            $this->checkSignal();
        }
    }

    public function fetchJob() {
        if ($this->fetchRows > 1) {
            $job = [];
            $i = 0;
            while ($i < $this->fetchRows) {
                $i++;
                if ($this->blockTimeout) {
                    $job[] = Core_Redis::getInstance(self::$connName)->blPop(self::$queueName, $this->blockTimeout);
                } else {
                    $job[] = Core_Redis::getInstance(self::$connName)->lPop(self::$queueName);
                }
            }
            $this->println("got jobs " . count($job));
        } else {
            if ($this->blockTimeout) {
                $job = Core_Redis::getInstance(self::$connName)->blPop(self::$queueName, $this->blockTimeout);
            } else {
                $job = Core_Redis::getInstance(self::$connName)->lPop(self::$queueName);
            }
            $this->println("got single job " . $job);
        }

        return $job;
    }

    public function redoJob($job) {
        if ($this->retryOnFailure) {
            $this->println("redo job " . serialize($job));
            $this->doJob($job);
        }
    }

    /**
     * @param mixed $job array or string
     */
    public static function doJob($job) {
        $job = (array)$job;
        foreach ($job as $j) {
            Core_Redis::getInstance(self::$connName)->lPush(self::$queueName, $j);
        }
    }

    abstract function worker($job);

}

